package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo09GroupBy {
  def main(args: Array[String]): Unit = {
    /**
     * groupBy：转换算子
     * 需要通过函数指定一个字段进行分组
     */
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo09GroupBy")

    val sc: SparkContext = new SparkContext(conf)
    val stuRDD: RDD[Stu] = sc
      .textFile("spark/data/stu/students.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        Stu(splits(0), splits(1), splits(2).toInt, splits(3), splits(4))
      })

    // 按照班级分组
    val stuGrpRDD: RDD[(String, Iterable[Stu])] = stuRDD.groupBy(_.clazz)
    stuGrpRDD.foreach(println)

    // 统计班级人数
    stuGrpRDD
      .map(kv => s"${kv._1},${kv._2.size}")
      .foreach(println)

  }
  case class Stu(id: String, name: String, age: Int, gender: String, clazz: String)

}
